/*
 * Copyright (C) 2018 The Asiainfo-Ability Authors
 *
 *      Licensed under the Apache License, Version 2.0 (the "License");
 *      you may not use this file except in compliance with the License.
 *      You may obtain a copy of the License at
 *
 *          http://www.apache.org/licenses/LICENSE-2.0
 *
 *      Unless required by applicable law or agreed to in writing, software
 *      distributed under the License is distributed on an "AS IS" BASIS,
 *      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *      See the License for the specific language governing permissions and
 *      limitations under the License.
 */

package org.asiainfo.ability.manager.service;


import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.asiainfo.ability.base.utils.ExceptionUtils;
import org.asiainfo.ability.base.utils.StringUtils;
import org.asiainfo.ability.gateway.code.FollowUpState;
import org.asiainfo.ability.base.code.InstanceState;
import org.asiainfo.ability.gateway.dao.ServiceDao;
import org.asiainfo.ability.gateway.model.Instance;
import org.asiainfo.ability.gateway.model.Machine;
import org.asiainfo.ability.manager.dao.NodeDao;
import org.asiainfo.ability.manager.dao.PackageDao;
import org.asiainfo.ability.manager.model.JarPackage;
import org.asiainfo.ability.manager.model.Node;
import org.asiainfo.ability.node.code.OperationType;
import org.asiainfo.ability.node.vo.NodeOperation;
import org.elasticsearch.common.inject.internal.InternalFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author: visen
 * @Date: 2018/2/1
 * @Description:
 */
@Service
public class InstanceService {

    private static final Logger logger = LogManager.getLogger(InstanceService.class);
    @Autowired
    private ServiceDao serviceDao;

    @Autowired
    private NodeDao nodeDao;

    @Autowired
    private SeqService seqService;

    @Autowired
    private Transaction transaction;

    @Autowired
    private PackageDao packageDao;

    @Autowired
    private PackageService packageService;

    @Autowired
    private AgentService agentService;

    private final static List<Instance> instanceQueue;
    private final static Set<String> nodeQueue;
    private final static ThreadPoolExecutor executor;

    static {
        instanceQueue = Collections.synchronizedList(new ArrayList<>());
        nodeQueue = Collections.synchronizedSet(new LinkedHashSet<>());
        executor = new ThreadPoolExecutor(2, 8, 300L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(65536));
    }

    @Scheduled(fixedDelay = 30000)
    public void queueTask() {

        if (!instanceQueue.isEmpty()) {
            for (Instance instance : instanceQueue) {
                nodeQueue.add(instance.getMachineId() + "." + instance.getRepoId());
            }
        }
    }

    public int updateInstance(Instance instance) {
        try {
            serviceDao.updateInstance(instance);
            return 1;
        } catch (Exception e) {
            return 0;
        }
    }
    private void batchProcess(List<Instance> instances, OperationType action) {
        //logger.error(transList);
        if (!instances.isEmpty()) {
            instanceQueue.removeAll(instances);
            if (action == OperationType.shutdown) {
                for (int i = 0, len = instances.size(); i < len; i++) {
                    Instance instance = instances.get(i);
                    instance.setState(InstanceState.shutdown);
                    serviceDao.updateInstance(instance);
                }
            } else {
                List<Instance> batchList = new ArrayList();
                for (int i = 0, len = instances.size(); i < len; ) {
                    batchList.add(instances.get(i));
                    if (++i % 5 == 0 || i == len) {
                        AgentService.BatchOperation batchOperation = new AgentService.BatchOperation();
                        batchOperation.setAction(action);
                        List<String> ids = new ArrayList();
                        batchOperation.setData(ids);
                        for (Instance instance : batchList) {
                            ids.add(instance.getId());
                        }
                        List<String> result = null;
                        try {
                            result = agentService.processNode(batchOperation);
                        } catch (Exception e) {
                            ExceptionUtils.logTrace(logger, e);
                        }

                        if (result != null) {
                            for (int j = 0, k = result.size(); j < k; j++) {
                                Instance instance = batchList.get(j);
                                instance.setState("1".equals(result.get(j)) ? (action == OperationType.start?
                                        InstanceState.active:InstanceState.pause) : InstanceState.shutdown);
                                updateInstance(instance);
                            }
                        }
                        batchList.clear();
                    }
                }
            }


        }

    }

    @Scheduled(fixedDelay = 30000)
    public void nodeQueueTask() {
        List<String> tempNodeQueue = new ArrayList(nodeQueue);
        for (String key : tempNodeQueue) {
            int i = key.indexOf(".");
            String machineId = key.substring(0, i);
            String repoId = key.substring(i + 1);

            executor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        long machinIdLong = Long.parseLong(machineId);
                        long repoIdLong = Long.parseLong(repoId);
                        Node node = nodeDao.getNode(machinIdLong, repoIdLong);
                        if (node == null) {

                            if (!packageService.dispatchPackage(packageDao.loadPackage(repoIdLong)
                                    , serviceDao.loadMachine(machinIdLong))) {
                                return;
                            }
                            node = new Node();
                            node.setMachineId(machinIdLong);
                            node.setRepoId(repoIdLong);
                            node.setUpdateTime(new Date());
                            nodeDao.saveNode(node);
                        }

                        List<Instance> tempInstanceQueue = new ArrayList(instanceQueue);
                        List<Instance> batchStartList = new ArrayList();
                        List<Instance> batchCloseList = new ArrayList();
                        List<Instance> batchSilentList = new ArrayList();
                        for (Instance item : tempInstanceQueue) {
                            if (item.getMachineId().equals(machineId) && item.getRepoId().equals(repoId)) {
                                if (item.getFollowUpState() == FollowUpState.start) {
                                    batchStartList.add(item);
                                } else if (item.getFollowUpState() == FollowUpState.shutdown) {
                                    batchCloseList.add(item);
                                } else if (item.getFollowUpState() == FollowUpState.silent) {
                                    batchSilentList.add(item);
                                }
                            }
                        }
                        batchProcess(batchStartList, OperationType.start);
                        batchProcess(batchCloseList, OperationType.shutdown);
                        batchProcess(batchSilentList, OperationType.silentStart);
                        nodeQueue.remove(key);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }
            });
        }


    }


    public static void main(String[] args) {
        List a = new ArrayList();
        a.add(1);
        a.add(2);
        a.add(3);
        System.out.println(a.subList(1, 3));
    }


    @Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.DEFAULT, timeout = 60000
            , rollbackFor = Exception.class)
    public List<Instance> scaleInstance(Instance instance, int initialCount) {
        JarPackage jarPackage = null;
        if (StringUtils.isBlank(instance.getRepoId())) {
            jarPackage = packageDao.getLatestMasterPackage(instance.getServiceName());
        } else {
            jarPackage = packageDao.loadPackage(Long.parseLong(instance.getRepoId()));
        }

        if (jarPackage == null) {
            throw new IllegalArgumentException("the jar package not exists");
        }
        instance.setBranch(jarPackage.getBranch());
        instance.setVersion(jarPackage.getVersion());
        instance.setRepoId(jarPackage.getId());
        //instance
        List<Machine> machines = serviceDao.listMachines();
        List<Instance> result = new ArrayList();
        Map fileters = new HashMap();
        try {
            for (; initialCount > 0; initialCount--) {
                Machine balcanceMachine = voteMachine(machines);
                Object maxPort = serviceDao.getMaxPort(Long.parseLong(balcanceMachine.getId()));

                int curPort = maxPort == null ? 23001 : ((Integer) maxPort + 1);
                if (curPort < 23001) curPort = 23001;
                Instance ins = new Instance();
                ins.setId(String.valueOf(seqService.next(SeqService.SeqTypes.INSTANCE)));
                ins.setServiceName(instance.getServiceName());
                ins.setRepoId(instance.getRepoId());
                ins.setBranch(instance.getBranch());
                ins.setVersion(instance.getVersion());
                ins.setMachineId(balcanceMachine.getId());
                ins.setMachineInfo(balcanceMachine.getName());
                ins.setWeight(instance.getWeight());
                ins.setState(InstanceState.prepare);
                ins.setPort(curPort++);
                ins.setFollowUpState(instance.getFollowUpState());
                instanceQueue.add(ins);
                /*
                String key = instance.getRepoId() + balcanceMachine.getId();
                Boolean needPrepare = (Boolean)fileters.get(key);
                if(needPrepare == null) {
                    needPrepare = checkIsReady(instance.getRepoId(), balcanceMachine.getId());
                    fileters.put(key, needPrepare);
                }
                if(needPrepare) {
                    instanceQueue.add(ins);
                }
                */

                ins.setMachineAddr(balcanceMachine.getIpAddress());
                serviceDao.saveInstance(ins);
                result.add(ins);
            }
        } catch (Exception e) {
            throw e;
        } finally {

        }


        return result;
    }


    public List<Instance> manualScaleInstance(Instance instance) {
        List<Instance> result = new ArrayList();
        instance.setId(String.valueOf(seqService.next(SeqService.SeqTypes.INSTANCE)));
        long machineId = Long.parseLong(instance.getMachineId());
        Machine machine = serviceDao.loadMachine(machineId);
        int usedFlag = serviceDao.isPortAlreadyUsed(machineId, instance.getPort());
        if (usedFlag > 0) {
            throw new IllegalArgumentException("port:" + instance.getPort() + " is already used");
        }
        JarPackage jarPackage = null;
        if (StringUtils.isBlank(instance.getRepoId())) {
            jarPackage = packageDao.getLatestMasterPackage(instance.getServiceName());
        } else {
            jarPackage = packageDao.loadPackage(Long.parseLong(instance.getRepoId()));
        }

        if (jarPackage == null) {
            throw new IllegalArgumentException("the jar package not exists");
        }
        instance.setBranch(jarPackage.getBranch());
        instance.setVersion(jarPackage.getVersion());
        instance.setRepoId(jarPackage.getId());
        instance.setState(InstanceState.prepare);
        instanceQueue.add(instance);
        serviceDao.saveInstance(instance);
        result.add(instance);
        instance.setMachineAddr(machine.getIpAddress());
        return result;
    }

    public Machine voteMachine(List<Machine> machines) {
        Machine machine = machines.get(0);
        return machine;
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true,
            isolation = Isolation.DEFAULT, timeout = 16000, rollbackFor = Exception.class)
    public String getInsCountSummary(String serviceId) throws Exception {
        int count = serviceDao.getInsCount(serviceId);
        if (count > 0) {
            return serviceDao.getActiveInsCount(serviceId) + " / " + count;
        } else {
            return "0 / 0";
        }
    }


    @Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.DEFAULT, timeout = 16000
            , rollbackFor = Exception.class)
    public void deleteInstances(String[] idArr) throws Exception {
        for (int i = 0, len = idArr.length; i < len; i++) {
            serviceDao.deleteInstance(Long.parseLong(idArr[i]));
        }
    }


}
